#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <signal.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <netinet/in.h>
#include <netdb.h>
#include "list.h"
#include "poller_protocol.h"

#define MAXPOLL 1024
#define perror2(s, e) fprintf(stderr, "%s: %s\n", s, strerror(e))

typedef struct request Request;

struct request
{
	int fd;
	time_t arrivalTime;
	time_t dispatchTime;
};

List pollData;
Request *requests;
int requestCount = -1, bufferSize;
bool terminate = false;

pthread_cond_t full = PTHREAD_COND_INITIALIZER;
pthread_cond_t empty = PTHREAD_COND_INITIALIZER;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

/* Public, read-only data */

int portnum, sock;
char hotIssueFileName[125];
int *requestsServed;
int workingThreads;
char  pollLog[125];
char statFile[125];

void Produce ( int newsock, long int arrivalTime );
void *Consume ( void *argument);
void Exit ( void );
bool GetVote ( char *pollAnswer );
void Terminate ( int signal );

int main ( int argc, char *argv[] )
{
	struct sockaddr_in server, client;
	struct sockaddr *serverptr, *clientptr;
	struct hostent *rem;
	int  newsock;
	int serverlen, clientlen;
	int i;
	pthread_t *threads;

	/*Check arguments */

	if ( argc < 6 )
	{
		fprintf(stderr, "Not enough arguments");
		return EXIT_FAILURE;
	}

	/* Get port number */

	portnum = atoi(argv[1]);

	/* Get number of working threads */

	workingThreads = atoi(argv[2]);

	if ( workingThreads <= 0 )
	{
		fprintf ( stderr, "Number of working threads must be positive number\n" );
		return 	EXIT_FAILURE;
	}

	threads = malloc ( sizeof(pthread_t) * workingThreads );
	requestsServed = malloc ( sizeof(int) * workingThreads );
	for ( i = 0; i < workingThreads; i++ )
	{
		requestsServed[i] = 0;
	}

	/* Get buffer size */

	bufferSize = atoi(argv[3]);

	if ( bufferSize <= 0 )
	{
		fprintf ( stderr, "Buffer size must be positive number\n" );
		return 	EXIT_FAILURE;
	}

	requests = malloc ( sizeof(Request) * bufferSize );	

	/* Get hot issue filename */

	strcpy ( hotIssueFileName, argv[4] );
	FILE *hotFile = fopen ( hotIssueFileName, "r" );

	if ( hotFile == NULL )
	{
		fprintf ( stderr, "%s does not exist\n", hotIssueFileName );
		exit ( 1 );	
	}
	fclose ( hotFile );


	/* Get poll log filename */

	strcpy ( pollLog, argv[5] );
	pollData = Create ( pollLog );
	sprintf ( statFile, "%s.stat", pollLog );

	if ( ( sock = socket(PF_INET, SOCK_STREAM, 0 ) ) < 0 )
	{
		perror ( "socket" );
		return EXIT_FAILURE;
	}


	/* Enable reuse of address */

	int optval = 1;
	setsockopt ( sock, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval) );

	server.sin_family = PF_INET;
	server.sin_addr.s_addr = htonl(INADDR_ANY);
	server.sin_port = htons(portnum);
	serverptr = (struct sockaddr *) &server;
	serverlen = sizeof(server);

	if ( bind ( sock, serverptr, serverlen ) < 0 )
	{
		perror("bind");
		return EXIT_FAILURE;
	}

	if ( listen ( sock, 5 ) < 0 )
	{
		perror ( "listen" );
		return EXIT_FAILURE;
	}

	signal ( SIGTERM, Terminate );

	/* Raise the working threads */

	for ( i = 0; i < workingThreads; i++ )
	{
		int error;
		if( error = pthread_create ( &threads[i], NULL, Consume, (void *) &i ) )
		{
			perror2("pthread_create", error);
		}
	}
	
	printf ( "Listening to connections at portnum %d\n", portnum );
	printf ( "My pid: %d ( In order to terminate run kill %d )\n", getpid(), getpid() );
	
	while ( true )
	{
		clientptr = (struct sockaddr *) &client;
		clientlen = sizeof(client);
		if ( ( newsock = accept ( sock, clientptr, &clientlen ) ) < 0 )
		{
			fprintf(stderr, "Connection closed\n" );
			break;
		}
		if ( ( rem = gethostbyaddr((char *) &client.sin_addr.s_addr, sizeof(client.sin_addr.s_addr ),
			client.sin_family ) ) == NULL )
		{
			perror("gethostbyaddr");
			return;
		}

		struct timeval tim;
		gettimeofday ( &tim, NULL );

		/* Place new request socket in the queue */

		Produce ( newsock, tim.tv_usec );
	}


	/* Unlock waiting threads */

	for ( i = 0; i < workingThreads; i++ )
	{
		pthread_cond_signal ( &empty );
	}

	for ( i = 0; i < workingThreads; i++ )
	{
		int err;

		if (err = pthread_join(threads[i], NULL ) ) 
		{
			perror2("pthread_join", err);		
		}	
	}

	Exit();
}

void Produce ( int newsock, long int arrivalTime )
{
	/* Thread enters Critical Section */

	pthread_mutex_lock( &mutex );

	/* If the queue is full, lock in the full condition
	 *variable, until a consumer thread unlocks it
	 */

	if ( requestCount == bufferSize - 1 )
	{
		pthread_cond_wait ( &full, &mutex );
	}

	requestCount++;
	requests[requestCount].fd =  newsock;
	requests[requestCount].arrivalTime = arrivalTime;

	/* If the queue qas empty, unlock the locked
	 *waiting consumer threads.
	 */

	if ( requestCount == 0 )
	{
		pthread_cond_signal ( &empty );
	}

	/* Thread enters Critical Section */

	pthread_mutex_unlock ( &mutex );
}

void *Consume ( void *argument )
{
	Request request;
	int threadID = * (int *) argument;
	char menu[] = "What would you like to do?\nChoice 1: Vote on Issue of the Day\nChoice 2: See Poll Results Collected thus far\nChoice 3: Quit\nEnter your choice:";

	while ( true )
	{
		/* Thread enters Critical Section */

		pthread_mutex_lock(&mutex);

		if ( requestCount == -1 )
		{
			pthread_cond_wait(&empty, &mutex);
		}

		if ( terminate == true )
		{
			pthread_mutex_unlock(&mutex);
			pthread_exit ( NULL );
		}

		request = requests[requestCount];
		int newSocket = request.fd;

		requestCount--;

		/* If the queue was full, unlock the blocked producer */

		if ( requestCount == bufferSize - 2)
		{
			pthread_cond_signal(&full);
		}	

		/* Thread leaves Critical Section */
		
		pthread_mutex_unlock(&mutex);

		/* Handle the request */
		int header;
		char message[MESSAGE_SIZE];
INITIAL_STATE:
		if ( Send ( MENU, menu, newSocket ) == false )
		{
			goto NEXT_REQUEST;
		}

		if ( WaitResponse ( &header, message, newSocket ) == false )
		{
			goto NEXT_REQUEST;
		}

		switch ( header )
		{
			bool exists;
			char username[MESSAGE_SIZE];

			case VOTE_BEGIN:
				if ( Send ( CREDENTIALS_REQUEST, "Please, type your username: ", newSocket ) == false )
				{
					goto NEXT_REQUEST;
				}

				if ( Wait ( CREDENTIALS_RESPONSE, username, newSocket ) == false )
				{
					goto NEXT_REQUEST;
				}				
				
				/* Search if the username exists in poll log */

				exists = Exists ( pollData, username );

				if ( exists == false )
				{
					/* Send the poll */

					FILE *poll = fopen ( hotIssueFileName, "r" );
					memset ( message, '\0', sizeof(message) );
					fread ( &message, sizeof(char), MAXPOLL, poll );
					fclose ( poll );		

					if ( Send ( VOTE_SUBJECT, message, newSocket ) == false )
					{
						goto NEXT_REQUEST;
					}

					/* Read the answer */

					if ( Wait ( VOTE, message, newSocket ) == false )
					{
						goto NEXT_REQUEST;
					}

					/* Submit the vote */

					bool vote = GetVote ( message );

					/* Enter Critical Section */

					pthread_mutex_lock( &mutex );

					Insert ( pollData,  username, vote );

					/* Leave Critical Section */
					
					pthread_mutex_unlock(&mutex);

					if ( Send ( VOTE_REGISTERED, "Your vote was submitted successfuly!\n", newSocket ) == false )
					{
						goto NEXT_REQUEST;
					}
					goto INITIAL_STATE;
				}
				else
				{
					/* If the username exists send message that the user
					 *cannot vote again.
					 */

					if ( Send ( ERROR_VOTE_AGAIN, "Sorry, you have already participated in the poll. You cannot vote 														again\n", newSocket ) == false )
					{
						goto NEXT_REQUEST;
					}
					goto INITIAL_STATE;
				}
				break;
			case POLL_RESULTS_BEGIN:
				if ( Send ( CREDENTIALS_REQUEST, "Please, type your username: ", newSocket ) == false )
				{
					goto NEXT_REQUEST;
				}

				if ( Wait ( CREDENTIALS_RESPONSE, username, newSocket ) == false )				
				{
					goto NEXT_REQUEST;
				}
				
				/* Search if the username exists in poll log */

				exists = Exists ( pollData, username );

				if ( exists == false )
				{
					if ( Send ( ERROR_NOT_VOTED, "You cannot see the results without first voting\n", newSocket ) == 																	false )
					{
						goto NEXT_REQUEST;
					}
					goto INITIAL_STATE;
				}
				else
				{
					/* Send the poll results*/	

					PollResults pollResults;
				 	GetPollResults ( pollData, &pollResults );
					sprintf ( message, "Yes: %d NO: %d", pollResults.yes, pollResults.no );
					if ( Send ( POLL_RESULTS, message, newSocket ) == false )
					{
						goto NEXT_REQUEST;
					}
					goto INITIAL_STATE;
				}
				
				break;
			case QUIT_CONNECTION:
				goto EXIT;
			default :
				break;
		}

		struct timeval time;
EXIT:	
		gettimeofday ( &time, NULL );
		request.dispatchTime = time.tv_usec - request.arrivalTime;
		requestsServed[threadID]++;

		/* Store request's arrival and dispatch time */

		FILE *file = fopen ( statFile, "a" );
		fprintf ( file, "Arrived %ld Dispatched %ld\n", request.arrivalTime, request.dispatchTime );
		fclose ( file );
NEXT_REQUEST:
		close ( newSocket );
	}
}

void Terminate ( int signal )
{
	/* Enter Critical Section */

	pthread_mutex_lock( &mutex );

	terminate = true;

	/* Leave Critical Section */
		
	pthread_mutex_unlock(&mutex);

	close ( sock );
}
void Exit ( void )
{	
	Flush ( pollData );
	
	/* Close the fds in the buffer */

	int i;
	for ( i = 0; i < requestCount; i++ )
	{
		close ( requests[i].fd );
	}

	/* Flush thread statistics */

	FILE *file  = fopen ( statFile, "a" );
	
	for ( i = 0; i < workingThreads; i++ )
	{		
		fprintf ( file, "Thread %d served %d requests\n", i, requestsServed[i] );
	}

	fclose ( file );
	close ( portnum );	
	exit ( 0 );
}


bool GetVote ( char *answer )
{
	return  strcmp ( answer, "YES" ) == 0 || strcmp ( answer, "yes" ) == 0;
}












